package org.apache.solr.ocean.replication;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.SortedMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.xml.crypto.Data;

import org.apache.solr.ocean.replication.RawLogFile.FileStreamRecord;
import org.apache.solr.ocean.replication.RawLogFile.RecordHeader;
import org.apache.solr.ocean.replication.RawLogFile.StreamRecord;
import org.apache.solr.ocean.util.ConcurrentTreeMap;

public class LogFile2 {
  private ConcurrentHashMap<String,ConcurrentTreeMap<Long,RecordHeader>> categorySnapshotMap = new ConcurrentHashMap<String,ConcurrentTreeMap<Long,RecordHeader>>();
  private ConcurrentTreeMap<Long,RecordHeader> snapshotMap = new ConcurrentTreeMap<Long,RecordHeader>();
  private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  private RawLogFile rawLogFile;
  private Long id;

  public LogFile2(Long id, File file) throws IOException {
    this.id = id;
    rawLogFile = new RawLogFile(file);
    List<RecordHeader> recordHeaders = rawLogFile.loadRecordHeaders();
    for (RecordHeader recordHeader : recordHeaders) {
      addRecordHeader(recordHeader);
    }
  }

  /**
   * public boolean checksum(String category, Long lowId, Long highId,
   * MessageDigest messageDigest) { ConcurrentTreeMap<Long, RecordHeader>
   * sortedMap = categorySnapshotMap.get(category); NavigableSet<Long> set =
   * sortedMap.subMap(lowId, true, highId, true).navigableKeySet(); return
   * checksum(messageDigest, set); }
   * 
   * public boolean checksum(Long lowId, Long highId, MessageDigest
   * messageDigest) { NavigableSet<Long> set = snapshotMap.subMap(lowId, true,
   * highId, true).navigableKeySet(); return checksum(messageDigest, set); }
   * 
   * public boolean checksum(MessageDigest messageDigest, Set<Long> set) {
   * byte[] buffer = new byte[8]; for (Long key : set) { NumberUtils.encode(key,
   * buffer); messageDigest.digest(buffer); } return true; }
   */
  public long size() {
    return rawLogFile.size();
  }

  public boolean delete(Long id) throws IOException {
    lock.writeLock().lock();
    try {
      RecordHeader recordHeader = snapshotMap.get(id);
      boolean deleted = rawLogFile.delete(recordHeader);
      if (deleted) {
        snapshotMap.remove(id);
        for (ConcurrentTreeMap<Long,RecordHeader> categoryMap : categorySnapshotMap.values()) {
          if (categoryMap.containsKey(id)) {
            categoryMap.remove(id);
          }
        }
      }
      return deleted;
    } finally {
      lock.writeLock().unlock();
    }
  }

  public Long getId() {
    return id;
  }

  public Long getPreviousId(Long id) {
    SortedMap<Long,RecordHeader> headMap = snapshotMap.headMap(id);
    return headMap.lastKey();
  }

  public Long getLowestId() {
    return snapshotMap.firstKey();
  }

  public boolean containsId(Long id) {
    return snapshotMap.containsKey(id);
  }

  public Long getHighestId() {
    List<Long> list = new LinkedList<Long>();
    for (ConcurrentTreeMap<Long,RecordHeader> map : categorySnapshotMap.values()) {
      Long lastKey = map.lastKey();
      if (lastKey != null)
        list.add(lastKey);
    }
    return Collections.max(list);
  }

  public boolean containsId(String category, Long id) {
    ConcurrentTreeMap<Long,RecordHeader> categoryMap = getCategoryMap(category);
    if (categoryMap == null)
      return false;
    return categoryMap.containsKey(id);
  }

  public List<RecordHeader> getRecordHeaders(Long fromKey, int num) {
    SortedMap<Long,RecordHeader> subMap = snapshotMap.tailMap(fromKey);
    Iterator<RecordHeader> iterator = subMap.values().iterator();
    int count = 0;
    List<RecordHeader> recordHeaders = new ArrayList<RecordHeader>(num);
    while (count < num && iterator.hasNext()) {
      RecordHeader recordHeader = iterator.next();
      recordHeaders.add(recordHeader);
      count++;
    }
    return recordHeaders;
  }

  public void commit(RecordHeader recordHeader) {
    addRecordHeader(recordHeader);
  }

  public RecordHeader writeRecord(Long id, String category, float schemaVersion, StreamRecord data) throws IOException {
    lock.writeLock().lock();
    try {
      RecordHeader recordHeader = rawLogFile.write(id, category, schemaVersion, data);
      addRecordHeader(recordHeader);
      return recordHeader;
    } finally {
      lock.writeLock().unlock();
    }
  }

  private void addRecordHeader(RecordHeader recordHeader) {
    snapshotMap.put(recordHeader.id, recordHeader);
    ConcurrentTreeMap<Long,RecordHeader> categoryMap = getCategoryMap(recordHeader.category);
    categoryMap.put(recordHeader.id, recordHeader);
  }
  
  public RecordIterator getRecordIterator(Long snapshotId) throws IOException {
    return new RecordIterator(snapshotId);
  }
  
  public class RecordIterator {
    private Iterator<RecordHeader> iterator;
    private RandomAccessFile randomAccessFile;
    
    public RecordIterator(Long snapshotId) throws IOException {
      File file = rawLogFile.getFile();
      randomAccessFile = new RandomAccessFile(file, "r");
      if (snapshotId == null) {
        iterator = snapshotMap.values().iterator();
      } else {
        iterator = snapshotMap.tailMap(snapshotId).values().iterator();
      }
    }
    
    public Record next() throws IOException {
      RecordHeader recordHeader = iterator.next();
      return getRecord(recordHeader);
    }
    
    private Record getRecord(RecordHeader recordHeader) throws IOException {
      FileStreamRecord fileStreamRecord = rawLogFile.readRecordData(recordHeader, randomAccessFile);
      Record record = new Record(recordHeader.id, recordHeader.category, recordHeader.schemaVersion, fileStreamRecord);
      return record;
    }
    
    public boolean hasNext() {
      return iterator.hasNext();
    }
    
    public void close() throws IOException {
      randomAccessFile.close();
    }
  }
  /**
  public class Reader {
    private RandomAccessFile randomAccessFile;
    
    public Reader() throws IOException {
      File file = rawLogFile.getFile();
      randomAccessFile = new RandomAccessFile(file, "r");
    }
    
    public void close() throws IOException {
      randomAccessFile.close();
    }
    
    public Record getRecordFromId(Long id) throws IOException {
      RecordHeader recordHeader = snapshotMap.getGreaterOrEqualTo(id);
      if (recordHeader == null)
        return null;
      return getRecord(recordHeader);
    }
    
    public List<Record> getRecords(String category, Long id, int num) throws {
      
    }
    
    public Record getRecord(RecordHeader recordHeader) throws IOException {
      FileStreamRecord fileStreamRecord = rawLogFile.readRecordData(recordHeader, randomAccessFile);
      Record record = new Record(recordHeader.id, recordHeader.category, recordHeader.schemaVersion, fileStreamRecord);
      return record;
    }

    public Record getRecordFromId(String category, Long id) throws IOException {
      ConcurrentTreeMap<Long,RecordHeader> map = categorySnapshotMap.get(category);
      RecordHeader recordHeader = map.getGreaterOrEqualTo(id);
      if (recordHeader == null)
        return null;
      return getRecord(recordHeader);
    }
  }
  **/
  public Long getLowestID(String category) {
    ConcurrentTreeMap<Long,RecordHeader> categoryMap = categorySnapshotMap.get(category);
    if (categoryMap == null)
      throw new IllegalArgumentException("category: " + category + " unknown");
    return categoryMap.firstKey();
  }

  public Long getHighestID(String category) {
    ConcurrentTreeMap<Long,RecordHeader> categoryMap = categorySnapshotMap.get(category);
    if (categoryMap == null)
      throw new IllegalArgumentException("category: " + category + " unknown");
    return categoryMap.lastKey();
  }

  public ConcurrentTreeMap<Long,RecordHeader> getCategoryMap(String category) {
    ConcurrentTreeMap<Long,RecordHeader> snapshotMap = categorySnapshotMap.get(category);
    if (snapshotMap == null) {
      snapshotMap = new ConcurrentTreeMap<Long,RecordHeader>();
      categorySnapshotMap.put(category, snapshotMap);
    }
    return snapshotMap;
  }

  public static class Record {
    private Long id;
    private String category;
    private float schemaVersion;
    private StreamRecord streamRecord;

    public Record(Long id, String category, float schemaVersion, StreamRecord streamRecord) {
      this.id = id;
      this.category = category;
      this.schemaVersion = schemaVersion;
      this.streamRecord = streamRecord;
    }

    public float getSchemaVersion() {
      return schemaVersion;
    }

    public String getCategory() {
      return category;
    }

    public Long getId() {
      return id;
    }

    public StreamRecord getStreamRecord() {
      return streamRecord;
    }
  }
}
